/*
 * To change this license header, choose License Headers in Project Properties.
 * To change this template file, choose Tools | Templates
 * and open the template in the editor.
 */
package cn.ac.iie.ban.data.exchange.runner.tokenBucket.server;

import cn.ac.iie.ban.data.exchange.runner.tokenBucket.exception.DEFCInitException;
import cn.ac.iie.ban.data.exchange.runner.tokenBucket.util.HttpHelper;
import cn.ac.iie.ban.data.exchange.runner.tokenBucket.util.Pair;
import cn.ac.iie.ban.data.exchange.runner.tokenBucket.util.PairList;
import cn.ac.iie.ban.data.exchange.runner.tokenBucket.util.ServiceInfo;
import cn.ac.iie.di.commons.speeder.lifecycle.LifeCycle;
import cn.ac.iie.di.commons.speeder.lifecycle.LifeCycleException;
import cn.ac.iie.di.commons.speeder.limiter.LazyTokenBucket;
import cn.ac.iie.di.commons.speeder.limiter.TokenBucket;
import cn.ac.iie.di.datadock.rdata.exchange.client.core.REFieldType;
import cn.ac.iie.di.datadock.rdata.exchange.client.exception.REConnectionException;
import cn.ac.iie.di.datadock.rdata.exchange.client.exception.RESessionException;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.connection.REConnection;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.session.RESendSession;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.session.RESendSessionBuilder;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.ReentrantLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import org.apache.log4j.Logger;

/**
 *
 * @author zwlin
 */
public class SpeedLimitHelper implements LifeCycle {
//
//    public static class Pair<K, V> {
//
//        public K key;
//        public V value;
//
//        public Pair(K key, V value) {
//            this.key = key;
//            this.value = value;
//        }
//
//    }

    private static final long DEFAULT_UPDATE_INTERVAL_SEC = 600L;

    private static final Logger LOGGER = Logger.getLogger(SpeedLimitHelper.class);

    //speed limiters
    private final TokenBucket totalBucket;
    //speed limit
    private final List<Integer> eachSpeed;
    private final List< Integer> eachSpeedIntevalSec;
    private final List<ConcurrentHashMap<String/*key*/, TokenBucket>> speedLimitList;
    private final Map<String, TokenBucket> specificLimitBuckets;
    //data size limit
    private final List<Integer> eachSize;
    private final List< Integer> eachSizeIntevalSec;
    private final List<ConcurrentHashMap<String/*key*/, TokenBucket>> sizeLimitList;

    //agency-uid map info keeper
    private HashMap<String/*uid*/, String/*key*/> keyMap;
    private final ReentrantLock updateLock;
    private long updateInterval;
    private long last_update;
    private final HttpHelper helper;
    private final Collection<String> metaUrls;
    private final String metaSuffix;

    //statistic info sender
    private int statisticInteval;
    private RESendSession sender;

    //timer to update and statistic
    private final Timer updateAndStatisticTimer;
    private final ReentrantReadWriteLock statisticLock;
    ConcurrentHashMap<String, AtomicInteger> statisticCache;

    //class state
    private boolean activated;

    public SpeedLimitHelper(ServiceInfo metaService,
            int totalSpeed, int eachSpeed)
            throws DEFCInitException {
        this(metaService, DEFAULT_UPDATE_INTERVAL_SEC, totalSpeed, eachSpeed);
    }

    public SpeedLimitHelper(ServiceInfo metaService, long updateIntervalSec,
            int totalSpeed, int eachSpeed)
            throws DEFCInitException {
        this(metaService, updateIntervalSec,
                totalSpeed, new PairList<Integer, Integer>(eachSpeed, 1));

    }

    public SpeedLimitHelper(
            ServiceInfo metaService,
            long updateIntervalSec,
            int totalSpeed,
            PairList<Integer /*max*/, Integer/*interval*/> speedControlPairList)
            throws DEFCInitException {
        this(metaService, updateIntervalSec, totalSpeed,
                speedControlPairList, new PairList<Integer, Integer>());
    }

    public SpeedLimitHelper(
            ServiceInfo metaService,
            long updateIntervalSec,
            int totalSpeed,
            PairList<Integer, Integer> speedList,
            PairList<Integer, Integer> dataSizeLimit
    ) throws DEFCInitException {
        this(metaService, updateIntervalSec,
                totalSpeed, speedList, dataSizeLimit, Collections.EMPTY_MAP);
    }

    public SpeedLimitHelper(
            ServiceInfo metaService,
            long updateIntervalSec,
            int totalSpeed,
            PairList<Integer, Integer> speedList,
            PairList<Integer, Integer> dataSizeLimit,
            Map<String, Integer> specificLimitMap
    ) throws DEFCInitException {
        //speed limiters
        this.totalBucket = new LazyTokenBucket(totalSpeed);
        this.speedLimitList = new ArrayList<>(speedList.size());
        this.eachSpeed = new ArrayList<>(speedList.size());
        this.eachSpeedIntevalSec = new ArrayList<>(speedList.size());
        speedList.forEach((Pair<Integer, Integer> pair) -> {
            this.speedLimitList.add(new ConcurrentHashMap<>());
            this.eachSpeed.add(pair.getKey());
            this.eachSpeedIntevalSec.add(pair.getValue());
        });

        this.sizeLimitList = new ArrayList<>(speedList.size());
        this.eachSize = new ArrayList<>(speedList.size());
        this.eachSizeIntevalSec = new ArrayList<>(speedList.size());
        dataSizeLimit.forEach((Pair<Integer, Integer> pair) -> {
            this.sizeLimitList.add(new ConcurrentHashMap<>());
            this.eachSize.add(pair.getKey());
            this.eachSizeIntevalSec.add(pair.getValue());
        });
        this.specificLimitBuckets = new HashMap<>();
        specificLimitMap.forEach((key, limit) -> {
            LazyTokenBucket bucket = new LazyTokenBucket(limit);
            bucket.start();
            this.specificLimitBuckets.put(key, bucket);
        });

        //agency-uid map and update
        this.keyMap = new HashMap<>();
        this.updateLock = new ReentrantLock();
        this.updateInterval = updateIntervalSec * 1000;
        this.last_update = 0;
        this.helper = new HttpHelper(false);
        this.metaUrls = metaService.getHosts();
        this.metaSuffix = metaService.getSuffix();

        //init statistic with none
        this.sender = null;
        this.statisticInteval = -1;
        this.statisticCache = null;
        this.statisticLock = new ReentrantReadWriteLock();

        //timer
        this.updateAndStatisticTimer = new Timer("update and statistic timer", true);

        //state not activated
        this.activated = false;
    }

    public void initStatistic(String namesrv, String topic, int intevalSec) throws REConnectionException, RESessionException, RESessionException {
        if (namesrv != null && !namesrv.isEmpty()
                && topic != null && !topic.isEmpty()
                && intevalSec >= 0) {
            REConnection conn = new REConnection(namesrv);
            RESendSessionBuilder builder = (RESendSessionBuilder) conn.getSendSessionBuilder(topic);
            builder.addColumn("scope", REFieldType.Int, false);
            builder.addColumn("uid", REFieldType.String, false);
            builder.addColumn("value", REFieldType.Int, false);
            this.sender = (RESendSession) builder.build();
            this.statisticInteval = intevalSec * 1000;
            this.statisticCache = new ConcurrentHashMap<>();
        }
    }

    @Override
    public void start() {
        if (!activated) {
            synchronized (this) {
                if (!activated) {
                    if (sender != null) {
                        try {
                            sender.start();
                        } catch (REConnectionException ex) {
                            LOGGER.fatal("statistic sender not started!sender is set null.", ex);
                            sender = null;
                        }
                    }
                    totalBucket.start();
                    updateAndStatisticTimer.schedule(new TimerTask() {
                        @Override
                        public void run() {
                            updateKeyMap();
                        }
                    }, 0, updateInterval);
                    if (statisticInteval > 0) {
                        updateAndStatisticTimer.schedule(new TimerTask() {
                            @Override
                            public void run() {
                                statistic();
                            }
                        }, 0, statisticInteval);
                    }
                    activated = true;
                }
            }
        }
    }

    @Override
    public void stop() {
        if (activated) {
            synchronized (this) {
                if (activated) {
                    activated = false;
                    updateAndStatisticTimer.cancel();
                    totalBucket.stop();
                    speedLimitList.forEach((ConcurrentHashMap<String, TokenBucket> map) -> {
                        for (TokenBucket bucket : map.values()) {
                            bucket.stop();
                        }
                    });
                    speedLimitList.clear();
                    keyMap.clear();
                    if (sender != null) {
                        statistic();
                        try {
                            sender.closeGracefully();
                        } catch (REConnectionException ex) {
                        }
                        try {
                            sender.shutdown();
                        } catch (REConnectionException ex) {
                        }
                    }
                    try {
                        helper.close();
                    } catch (IOException ex) {
                    }
                }
            }
        }
    }

    public boolean get(String uid, int num) throws LifeCycleException {
        String key = getKey(uid);
        if (key == null) {
            return false;
        }
        confirmKey(key);
        boolean rtn = totalBucket.get(num)
                && get(key, num, sizeLimitList)
                && get(key, num, speedLimitList)
                && (specificLimitBuckets.containsKey(key)
                ? specificLimitBuckets.get(key).get(num) : true);
        if (rtn) {
            statistic(uid, num);
        }
        return rtn;
    }

    public boolean tryGet(String uid, int num) throws LifeCycleException {
        String key = getKey(uid);
        if (key == null) {
            return false;
        }
        confirmKey(key);
        boolean rtn = totalBucket.get(num)
                && get(key, num, sizeLimitList)
                && tryGet(key, num, speedLimitList)
                && (specificLimitBuckets.containsKey(key)
                ? specificLimitBuckets.get(key).tryGet(num) : true);
        if (rtn) {
            statistic(uid, num);
        }
        return rtn;
    }

    private void confirmKey(String key) {
        for (int i = 0; i < speedLimitList.size(); ++i) {
            ConcurrentHashMap<String, TokenBucket> limit = speedLimitList.get(i);
            if (!limit.containsKey(key)) {
                synchronized (limit) {
                    if (!limit.containsKey(key)) {
                        TokenBucket bucket = new LazyTokenBucket(eachSpeed.get(i), eachSpeedIntevalSec.get(i) * 1000);
                        limit.put(key, bucket);
                        bucket.start();
                    }
                }
            }
        }
        for (int i = 0; i < sizeLimitList.size(); ++i) {
            ConcurrentHashMap<String, TokenBucket> limit = sizeLimitList.get(i);
            if (!limit.containsKey(key)) {
                synchronized (limit) {
                    if (!limit.containsKey(key)) {
                        TokenBucket bucket = new LazyTokenBucket(eachSize.get(i), eachSizeIntevalSec.get(i) * 1000);
                        limit.put(key, bucket);
                        bucket.start();
                    }
                }
            }
        }
    }

    public boolean get(String key, int size, List<ConcurrentHashMap<String/*key*/, TokenBucket>> limitList) throws LifeCycleException {
        for (ConcurrentHashMap<String, TokenBucket> map : limitList) {
            if (!map.get(key).get(size)) {
                return false;
            }
        }
        return true;
    }

    public boolean tryGet(String key, int size, List<ConcurrentHashMap<String/*key*/, TokenBucket>> limitList) throws LifeCycleException {
        for (ConcurrentHashMap<String, TokenBucket> map : limitList) {
            if (!map.get(key).tryGet(size)) {
                return false;
            }
        }
        return true;
    }

    private String getKey(String uid) {
        if (!keyMap.containsKey(uid)) {
            updateKeyMap();
        }
        return keyMap.get(uid);
    }

    private void updateKeyMap() {
        LOGGER.debug("starting update key map");
        if (System.currentTimeMillis() - last_update < 10000) {
            return;
        }
        updateLock.lock();
        try {
            if (System.currentTimeMillis() - last_update < 10000) {
                return;
            }
            HttpHelper.Result rslt = helper.get(metaUrls, metaSuffix);
            if (rslt.getStatusCode() != 200) {
                LOGGER.error("update key map error, ret code: " + rslt.getStatusCode()
                        + " , reason: " + rslt.getReasonPhrase()
                        + " , content: " + rslt.getContent());
                //error
            } else {
                String str = rslt.getContent();
                LOGGER.debug("update key map result: " + str);
                //handle
                JSONObject obj = JSONObject.parseObject(str);
                if (!obj.containsKey("status")) {
                    LOGGER.error("update key map error, no status in content: " + str);
                    return;
                }
                if (!obj.getString("status").trim().equalsIgnoreCase("success")) {
                    LOGGER.error("update key map error. not success: " + str);
                }
                JSONArray array = obj.getJSONArray("result");
                if (array == null || array.isEmpty()) {
                    LOGGER.warn("got empty map");
                    return;
                }
                HashMap<String, String> map = new HashMap<>();
                for (Object tupleObj : array) {
                    try {
                        JSONObject tuple = (JSONObject) tupleObj;
                        map.put(tuple.getString("id"), tuple.getString("agency__id"));
                    } catch (Exception e) {
                        LOGGER.error("unknow error appeared and skip current tuple. current : " + tupleObj + ", int content: " + str);
                    }
                }
                keyMap = map;
            }
        } catch (Throwable t) {
            LOGGER.error("update key map error.", t);
        } finally {
            last_update = System.currentTimeMillis();
            updateLock.unlock();
        }
    }

    private void statistic() {
        if (sender == null) {
            return;
        }
        ConcurrentHashMap<String, AtomicInteger> tmp = statisticCache;
        statisticLock.writeLock().lock();
        try {
            statisticCache = new ConcurrentHashMap<>();
        } finally {
            statisticLock.writeLock().unlock();
        }
        String uid;
        int num;
        itr:
        for (Map.Entry<String, AtomicInteger> entry : tmp.entrySet()) {
            uid = entry.getKey();
            num = entry.getValue().get();
            try {
                sender.setInt("scope", -1);
                sender.setString("uid", uid);
                sender.setInt("value", num);
            } catch (RESessionException ex) {
                LOGGER.warn("cannot add value.scope=-1,uid=" + uid + ",value=" + num, ex);
                continue;
            }
            for (int i = 0; i < 3; ++i) {
                try {
                    sender.add();
                    continue itr;
                } catch (RESessionException ex) {
                    LOGGER.warn("cannot add rec, trying to flush.", ex);
                    try {
                        sender.flush();
                    } catch (Exception ex1) {
                        LOGGER.warn("cannot flush data.");
                    }
                }
            }
            LOGGER.error("cannot add rec after 3 trails, give up.");
            return;
        }
        for (int i = 0; i < 3; ++i) {
            try {
                sender.flush();
                return;
            } catch (Exception ex) {
                LOGGER.warn("cannot flush data, retrying.", ex);
            }
        }
        LOGGER.error("cannot flush data after 3 trails, give up.");
    }

    private void statistic(String uid, int num) {
        if (sender == null) {
            return;
        }
        if (statisticInteval > 0) {
            statisticLock.readLock().lock();
            try {
                if (statisticCache.containsKey(uid)) {
                } else {
                    statisticCache.putIfAbsent(uid, new AtomicInteger());
                }
                statisticCache.get(uid).addAndGet(num);
            } finally {
                statisticLock.readLock().unlock();
            }
        } else {
            statisticLock.writeLock().lock();
            try {
                sender.setInt("scope", -1);
                sender.setString("uid", uid);
                sender.setInt("value", num);
                sender.add();
                sender.flush();
            } catch (RESessionException | REConnectionException ex) {
                LOGGER.error("send statistic info failed.", ex);
            } finally {
                statisticLock.writeLock().unlock();
            }
        }
    }

}
